package com.atguigu.bigdata.edu.realtime.app.dwd;

import com.atguigu.bigdata.edu.realtime.app.BaseSQLApp;
import com.atguigu.bigdata.edu.realtime.common.Constant;
import com.atguigu.bigdata.edu.realtime.util.SQLUtil;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.Table;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;
import org.apache.hadoop.yarn.webapp.hamlet2.Hamlet;

import java.time.Duration;


public class Dwd_03_DwdTradeOrderDetail extends BaseSQLApp {
    public static void main(String[] args) {
        new Dwd_03_DwdTradeOrderDetail().init(
                3030,
                2,
                "Dwd_03_DwdTradeOrderDetail"
        );
    }


    @Override
    protected void handle(StreamExecutionEnvironment env,
                          StreamTableEnvironment tEnv) {
        tEnv.getConfig().setIdleStateRetention(Duration.ofSeconds(5));

        // 1. 读取 ods_db
        readOdsDb(tEnv, "Dwd_03_DwdTradeOrderDetail");

        //读取source
        readBaseSource(tEnv);

        // 2. 过滤出详情表
        Table orderDetail = tEnv.sqlQuery("select " +
                "data['id'] id,  " +
                "data['order_id'] order_id,  " +
                "data['user_id'] user_id,  " +
                "data['course_id'] course_id,  " +
                "data['course_name'] course_name,  " +
                "data['origin_amount'] origin_amount,  " +
                "data['final_amount'] final_amount,  " +
                "data['create_time'] create_time, " +
                "ts,  " +
                "pt  " +
                "from ods_db " +
                "where `database`='edu' " +
                "and `table`='order_detail' " +
                "and `type`='insert' ");
        tEnv.createTemporaryView("order_detail", orderDetail);
//        orderDetail.execute().print();

        //读取common
        tEnv.executeSql("create table dwd_traffic_common( " +
                "uid bigint, " +
                "sc bigint, " +
                "sid string, " +
                "pt as proctime()" +
                ")"+ SQLUtil.getKafkaSource(Constant.TOPIC_DWD_TRAFFIC_COMMON, "Dwd_03_DwdTradeOrderDetail"));

        //tEnv.sqlQuery("select * from dwd_traffic_common").execute().print();

        // 4. 过滤订单表
        Table orderInfo = tEnv.sqlQuery("select " +
                "data['id'] id," +
                "data['user_id'] user_id," +
                "data['province_id'] province_id, " +
                "data['out_trade_no'] out_trade_no, " +
                "data['session_id']  session_id " +
                "from ods_db " +
                "where `database`='edu' " +
                "and `table`='order_info' " +
                "and `type`='insert' ");
        tEnv.createTemporaryView("order_info", orderInfo);



        // 7. join
        Table result = tEnv.sqlQuery("select " +
                "od.id,  " +
                "od.order_id,  " +
                "oi.user_id,  " +
                "od.course_id,  " +
                "od.course_name,  " +
                "oi.province_id,  " +
                "date_format(od.create_time, 'yyyy-MM-dd') date_id,  " +
                "od.create_time,  " +
                "od.origin_amount,  " +
                "od.final_amount,  " +
                "oi.session_id,  " +
                "oi.out_trade_no,  " +
                "src.id source_id, " +
                "src.source_site, " +
                "od.ts, " +
                "current_row_timestamp() row_op_ts  " + /*得到每行计算的时间: 到后面 dws 去重的时候使用*/
                "from order_detail od " +
                "join order_info oi on od.order_id=oi.id " +
                "join dwd_traffic_common tc on oi.session_id = tc.sid "+
                "join base_source for system_time as of od.pt as src on tc.sc = src.id");

        //result.execute().print();

        // 7. 写出到 kafka 中
        tEnv.executeSql("create table dwd_trade_order_detail(  " +
                "id string,  " +
                "order_id string,  " +
                "user_id string,  " +
                "course_id string,  " +
                "course_name string,  " +
                "province_id string,  " +
                "date_id string,  " +
                "create_time string,  " +
                "origin_amount string,  " +
                "final_amount string,  " +
                "session_id string,  " +
                "out_trade_no string, " +
                "source_id bigint," +
                "source_site string, " +
                "ts bigint,  " +
                "row_op_ts timestamp_ltz(3),  " +
                "primary key(id) not enforced  " +
                ")" + SQLUtil.getUpsertKafkaSink(Constant.TOPIC_DWD_TRADE_ORDER_DETAIL));



        result.executeInsert("dwd_trade_order_detail");
    }
}



